package com.mm.fam.node.service.impl;

import com.alibaba.fastjson.JSON;
import com.mm.fam.common.service.FastEventService;
import com.mm.fam.common.utils.CommonUtils;
import com.mm.fam.gateway.api.GatewayApi;
import com.mm.fam.node.rpc.service.RpcClientApiService;
import com.mm.fam.node.service.SlaveSystemService;
import com.mm.fam.pb.CoreEnum.ConnectStatusEnum;
import com.mm.fam.pb.CoreEnum.GatewayTypeEnum;
import com.mm.fam.pb.CoreField.ContractField;
import com.mm.fam.pb.CoreField.GatewayField;
import com.mm.fam.pb.CoreField.GatewaySettingField;
import com.mm.fam.pb.CoreRpc.RpcSyncSlaveNodeRuntimeDataRsp;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import java.lang.reflect.Constructor;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
public class SlaveSystemServiceImpl implements SlaveSystemService, InitializingBean {

    private static Logger logger = LoggerFactory.getLogger(SlaveSystemServiceImpl.class);

    //	@Autowired
//	private SlaveTradeCachesService slaveTradeCachesService;
    @Autowired
    private FastEventService fastEventService;

    @Autowired
    private RpcClientApiService rpcClientApiService;

    private Map<String, GatewayApi> gatewayApiMap = new HashMap<>(100);

    private Map<String, ContractField> subscribedContractMap = new HashMap<>();

    private ExecutorService executor = Executors.newCachedThreadPool();

//    @Value("${rt.rpc.client.node-id}")
//    private Integer nodeId;
//
//    @Value("${rt.node.slave.operatorId}")
//    private String slaveOperatorId;

    @Value("${rt.node.slave.sync-runtime-data-wait-seconds}")
    private Long syncRuntimeDataWaitSeconds;

    /**
     * 初始化后执行
     * 异步启动ctp入口
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        logger.info("com.mm.fam包下扫描到了以下GatewayApi实现类,{}", JSON.toJSONString(this.scanGatewayApiImpl()));
        executor.execute(new SyncSlaveNodeRuntimeDataTask(syncRuntimeDataWaitSeconds));
    }

    private List<String> scanGatewayApiImpl() {
        List<String> gatewayClassNameList = new ArrayList<>();
        Set<Class<?>> classes = CommonUtils.getClasses("com.mm.fam");
        if (classes == null) {
            logger.warn("com.mm.fam下扫描到任何类");
        } else {
            // 寻找Gateway的实现类,不包含抽象类
            Set<Class<?>> filteredClasses = CommonUtils.getImplementsByInterface(GatewayApi.class, classes, false);
            if (filteredClasses.isEmpty()) {
                logger.warn("com.mm.fam下扫描到任何Gateway网关的实现类");
            } else {
                for (Class<?> clazz : filteredClasses) {
                    String className = clazz.getName();
                    gatewayClassNameList.add(className);
                }
            }
        }
        return gatewayClassNameList;
    }

    @Override
    public List<String> getConnectedGatewayIdList() {
        List<String> resultList = new ArrayList<String>();

        for (Map.Entry<String, GatewayApi> entry : gatewayApiMap.entrySet()) {
            String gatewayId = entry.getKey();
            GatewayApi gatewayApi = entry.getValue();
            if (gatewayApi.isConnected()) {
                resultList.add(gatewayId);
            }
        }

        return resultList;
    }

    @Override
    public synchronized void connectGatewayApi(GatewaySettingField gatewaySetting) {
        String gatewayId = gatewaySetting.getGatewayId();
        String gatewayName = gatewaySetting.getGatewayName();
        logger.info("连接网关,网关ID:{},网关名称:{}", gatewayId, gatewayName);
        if (gatewayApiMap.containsKey(gatewayId)) {
            logger.warn("网关已在缓存中存在,网关ID:{},网关名称:{}", gatewayId, gatewayName);
            GatewayApi gatewayApi = gatewayApiMap.get(gatewaySetting.getGatewayId());
            if (gatewayApi.isConnected()) {
                logger.error("连接网关错误,网关ID:{},网关名称:{},缓存中的网关处于连接状态", gatewayId, gatewayName);
                return;
            } else {
                logger.warn("缓存中的网关已经断开,再次调用网关断开并删除,网关ID:{},网关名称:{}", gatewayId, gatewayName);
                disconnectGatewayApi(gatewayId);
            }
        }

        String gatewayClassName = gatewaySetting.getImplementClassName();
        try {
            logger.info("使用反射创建网关实例,网关ID:{},网关名称:{}", gatewayId, gatewayName);
            Class<?> clazz = Class.forName(gatewayClassName);
            Constructor<?> c = clazz.getConstructor(FastEventService.class, GatewaySettingField.class);
            GatewayApi gatewayApi = (GatewayApi) c.newInstance(fastEventService, gatewaySetting);
            logger.info("调用网关连接,网关ID:{},网关名称:{}", gatewayId, gatewayName);
            gatewayApi.connect();
            logger.info("重新订阅合约,网关ID:{},网关名称:{}", gatewayId, gatewayName);
            // 重新订阅之前的合约
            for (ContractField contract : subscribedContractMap.values()) {
                gatewayApi.subscribe(contract);
            }
            //缓存已启动的网关
            gatewayApiMap.put(gatewayId, gatewayApi);
        } catch (Exception e) {
            logger.error("连接网关错误,创建网关实例发生异常,网关ID:{},Java实现类:{}", gatewayId, gatewayClassName, e);
        }
        logger.warn("连接网关完成,网关ID:{},网关名称:{}", gatewayId, gatewayName);
    }

    @Override
    public synchronized void disconnectGatewayApi(String gatewayId) {
        GatewayApi gatewayApi = getGatewayApi(gatewayId);
        if (gatewayApi != null) {
            gatewayApi.disconnect();
            gatewayApiMap.remove(gatewayId);
//			slaveTradeCachesService.removeAllCachesByGatewayId(gatewayId);
        } else {
            logger.error("网关{}不存在,无法断开!", gatewayId);
        }
    }

    @Override
    public GatewayApi getGatewayApi(String gatewayId) {
        if (StringUtils.isEmpty(gatewayId)) {
            logger.error("获取网关,gatewayId不允许使用空字符串或null");
            return null;
        }
        return gatewayApiMap.get(gatewayId);
    }

    @Override
    public List<GatewayApi> getGatewayApiList() {
        return new ArrayList<>(gatewayApiMap.values());
    }

    class SyncSlaveNodeRuntimeDataTask implements Runnable {

        private long waitSeconds = 8;

        // 设置等待时间   与websocket 链接时间错开
        public SyncSlaveNodeRuntimeDataTask(long waitSeconds) {
            if (waitSeconds >= 8 && waitSeconds <= 300) {
                this.waitSeconds = waitSeconds;
            } else {
                logger.warn("使用默认等待时间配置");
            }

        }

        @Override
        public void run() {
            /**
             * 每50毫秒检查一次
             * 是否有新的网管数据传过来,有则联通
             */
            long startTime = System.currentTimeMillis();
            while (true) {
                if ((System.currentTimeMillis() - startTime) < waitSeconds * 1000) {
                    try {
                        Thread.sleep(50);
                        continue;
                    } catch (InterruptedException e) {
                        logger.error("定时同步节点数据等待检测到线程中断", e);
                        break;
                    }
                }

                try {
                    List<GatewayField> gatewayList = new ArrayList<>();

                    for (GatewayApi gatewayApi : getGatewayApiList()) {
                        String gatewayId = gatewayApi.getGatewayId();

                        ConnectStatusEnum gatewayStatus = ConnectStatusEnum.DISCONNECTED;
                        if (gatewayApi.isConnected()) {
                            gatewayStatus = ConnectStatusEnum.CONNECTED;
                        }
                        GatewayField gateway = GatewayField.newBuilder().setGatewayId(gatewayId).setStatus(gatewayStatus).build();
                        gatewayList.add(gateway);

                    }


                    // 获取网关数据(账户密码,行情交易地址等....)
                    RpcSyncSlaveNodeRuntimeDataRsp rpcSyncSlaveNodeRuntimeDataRsp = rpcClientApiService.syncSlaveNodeRuntimeData(gatewayList);

                    if (rpcSyncSlaveNodeRuntimeDataRsp != null) {
                        List<ContractField> newSubscribedContractList = rpcSyncSlaveNodeRuntimeDataRsp.getSubscribedContractList();

                        Set<String> newSubscribedUnifiedSymbolSet = new HashSet<>();

                        for (ContractField contract : newSubscribedContractList) {
                            newSubscribedUnifiedSymbolSet.add(contract.getUnifiedSymbol());
                        }

                        List<ContractField> unsubscribeContractList = new ArrayList<>();
                        for (Map.Entry<String, ContractField> entry : subscribedContractMap.entrySet()) {
                            String unifiedSymbol = entry.getKey();
                            ContractField contract = entry.getValue();
                            if (!newSubscribedUnifiedSymbolSet.contains(unifiedSymbol)) {
                                unsubscribeContractList.add(contract);
                            }
                        }

                        for (ContractField contract : unsubscribeContractList) {
                            subscribedContractMap.remove(contract.getUnifiedSymbol());
                            for (GatewayApi gatewayApi : gatewayApiMap.values()) {
                                if (gatewayApi.getGateway().getGatewayType().equals(GatewayTypeEnum.MARKET_DATA)
                                        || gatewayApi.getGateway().getGatewayType().equals(GatewayTypeEnum.TRADE_AND_MARKET_DATA)) {
                                    try {
                                        gatewayApi.unsubscribe(contract);
                                    } catch (Exception e) {
                                        logger.error("取消订阅发生异常", e);
                                    }
                                }
                            }
                        }

                        for (ContractField contract : newSubscribedContractList) {
                            if (!subscribedContractMap.containsKey(contract.getUnifiedSymbol())) {
                                subscribedContractMap.put(contract.getUnifiedSymbol(), contract);
                                for (GatewayApi gatewayApi : gatewayApiMap.values()) {
                                    if (gatewayApi.getGateway().getGatewayType().equals(GatewayTypeEnum.MARKET_DATA)
                                            || gatewayApi.getGateway().getGatewayType().equals(GatewayTypeEnum.TRADE_AND_MARKET_DATA)) {
                                        try {
                                            gatewayApi.subscribe(contract);
                                        } catch (Exception e) {
                                            logger.error("订阅发生异常", e);
                                        }
                                    }
                                }
                            }
                        }

                        List<GatewaySettingField> gatewaySettingList = rpcSyncSlaveNodeRuntimeDataRsp.getGatewaySettingList();
                        if (gatewaySettingList != null) {

                            Map<String, GatewaySettingField> gatewaySettingMap = new HashMap<>();
                            for (GatewaySettingField gatewaySetting : gatewaySettingList) {
                                gatewaySettingMap.put(gatewaySetting.getGatewayId(), gatewaySetting);
                            }

                            List<GatewayApi> gatewayApiList = getGatewayApiList();

                            // 断开连接
                            for (GatewayApi gatewayApi : gatewayApiList) {
                                String gatewayId = gatewayApi.getGatewayId();
                                GatewaySettingField gatewaySetting = gatewaySettingMap.get(gatewayId);
                                if (gatewaySetting == null) {
                                    disconnectGatewayApi(gatewayId);
                                } else {
                                    if (gatewaySetting.getStatus() == ConnectStatusEnum.DISCONNECTING
                                            || gatewaySetting.getStatus() == ConnectStatusEnum.DISCONNECTED) {
                                        logger.info("网关{}状态变更,执行断开操作", gatewayId);
                                        disconnectGatewayApi(gatewayId);
                                    } else if (gatewaySetting.getVersion() != gatewayApi.getGatewaySetting().getVersion()) {
                                        logger.info("网关{}配置变更,执行断开操作", gatewayId);
                                        disconnectGatewayApi(gatewayId);
                                    }
                                }

                            }

                            //启动 状态为连接中 已连接的并且没有启动的网关
                            for (GatewaySettingField gatewaySetting : gatewaySettingList) {
                                if (gatewaySetting.getStatus() == ConnectStatusEnum.CONNECTING || gatewaySetting.getStatus() == ConnectStatusEnum.CONNECTED) {

                                    GatewayApi gatewayApi = getGatewayApi(gatewaySetting.getGatewayId());
                                    if (gatewayApi == null || !gatewayApi.isConnected()) {
                                        logger.info("执行连接网关操作", gatewaySetting.getGatewayId());
                                        // 根据网管启动方法
                                        connectGatewayApi(gatewaySetting);
                                    }
                                }

                            }

                        }
                    } else {
                        logger.warn("定时同步节点数据,未能获取到数据");
                    }

                } catch (Exception e) {
                    logger.error("定时同步节点数据发生异常", e);
                }

                logger.info("定时同步节点数据完成");

                startTime = System.currentTimeMillis();
            }
        }

    }

}
